package com.bfxy.disruptor.common;

import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;

import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.consume.MutilConsumer;
import com.bfxy.disruptor.common.event.EventFactory;
import com.bfxy.disruptor.common.exception.ExceptionHandler;
import com.bfxy.disruptor.common.producer.Productor;
import com.bfxy.disruptor.common.waitStraty.BlockingWaitStrategy;

/**
 * WorkerPool contains a pool of {@link MutilConsumer}s that will consume
 * sequences so jobs can be farmed out across a pool of workers. Each of the
 * {@link MutilConsumer}s manage and calls a {@link WorkHandler} to process the
 * events.
 *
 * @param <T> event to be processed by a pool of workers
 */
//多消费者模式
//WorkerPool里面有消费者数组
public final class WorkerPool<T> {
	private final AtomicBoolean started = new AtomicBoolean(false);
	// 每一个消费者的序号，初始值是-1，
	private final Sequence workSequence = new Sequence(Productor.INITIAL_CURSOR_VALUE);
	private final RingBuffer<T> ringBuffer;
	// WorkProcessors are created to wrap each of the provided WorkHandlers
	private final MutilConsumer<?>[] workProcessors;

	/**
	 * Create a worker pool to enable an array of {@link WorkHandler}s to consume
	 * published sequences.
	 *
	 * This option requires a pre-configured {@link RingBuffer} which must have
	 * {@link RingBuffer#addGatingSequences(Sequence...)} called before the work
	 * pool is started.
	 *
	 * @param ringBuffer       of events to be consumed.
	 * @param sequenceBarrier  on which the workers will depend.
	 * @param exceptionHandler to callback when an error occurs which is not handled
	 *                         by the {@link WorkHandler}s.
	 * @param workHandlers     to distribute the work load across.
	 */

	public WorkerPool(final RingBuffer<T> ringBuffer, final SequenceBarrier sequenceBarrier,
			final ExceptionHandler<? super T> exceptionHandler, final WorkHandler<? super T>... workHandlers) {
		this.ringBuffer = ringBuffer;
		final int numWorkers = workHandlers.length;
		// 一个消费者就是一个WorkProcessor，一个消费者一个workSequence序号，
		workProcessors = new MutilConsumer[numWorkers];
		// 多少个消费者就有多少个WorkProcessor，workHandlers就是消费者数组。
		for (int i = 0; i < numWorkers; i++) {
			workProcessors[i] = new MutilConsumer<T>(ringBuffer, sequenceBarrier, workHandlers[i], exceptionHandler,
					workSequence);
		}
	}

	/**
	 * Construct a work pool with an internal {@link RingBuffer} for convenience.
	 *
	 * This option does not require
	 * {@link RingBuffer#addGatingSequences(Sequence...)} to be called before the
	 * work pool is started.
	 *
	 * @param eventFactory     for filling the {@link RingBuffer}
	 * @param exceptionHandler to callback when an error occurs which is not handled
	 *                         by the {@link WorkHandler}s.
	 * @param workHandlers     to distribute the work load across.
	 */
	public WorkerPool(final EventFactory<T> eventFactory, final ExceptionHandler<? super T> exceptionHandler,
			final WorkHandler<? super T>... workHandlers) {
		ringBuffer = RingBuffer.createMultiProducer(eventFactory, 1024, new BlockingWaitStrategy());
		final SequenceBarrier barrier = ringBuffer.newBarrier();
		final int numWorkers = workHandlers.length;
		workProcessors = new MutilConsumer[numWorkers];

		for (int i = 0; i < numWorkers; i++) {
			workProcessors[i] = new MutilConsumer<T>(ringBuffer, barrier, workHandlers[i], exceptionHandler,
					workSequence);
		}

		ringBuffer.addGatingSequences(getWorkerSequences());
	}

	/**
	 * Get an array of {@link Sequence}s representing the progress of the workers.
	 *
	 * @return an array of {@link Sequence}s representing the progress of the
	 *         workers.
	 */
	public Sequence[] getWorkerSequences() {
		final Sequence[] sequences = new Sequence[workProcessors.length + 1];
		for (int i = 0, size = workProcessors.length; i < size; i++) {
			sequences[i] = workProcessors[i].getSequence();
		}
		sequences[sequences.length - 1] = workSequence;

		return sequences;
	}

	/**
	 * Start the worker pool processing events in sequence.
	 *
	 * @param executor providing threads for running the workers.
	 * @return the {@link RingBuffer} used for the work queue.
	 * @throws IllegalStateException if the pool has already been started and not
	 *                               halted yet
	 */
	public RingBuffer<T> start(final Executor executor) {
		if (!started.compareAndSet(false, true)) {
			throw new IllegalStateException(
					"WorkerPool has already been started and cannot be restarted until halted.");
		}

		final long cursor = ringBuffer.getCursor();
		workSequence.set(cursor);

		for (MutilConsumer<?> processor : workProcessors) {
			processor.getSequence().set(cursor);
			executor.execute(processor);
		}

		return ringBuffer;
	}

	/**
	 * Wait for the {@link RingBuffer} to drain of published events then halt the
	 * workers.
	 */
	public void drainAndHalt() {
		Sequence[] workerSequences = getWorkerSequences();
		while (ringBuffer.getCursor() > Util.getMinimumSequence(workerSequences)) {
			Thread.yield();
		}

		for (MutilConsumer<?> processor : workProcessors) {
			processor.halt();
		}

		started.set(false);
	}

	/**
	 * Halt all workers immediately at the end of their current cycle.
	 */
	public void halt() {
		for (MutilConsumer<?> processor : workProcessors) {
			processor.halt();
		}

		started.set(false);
	}

	public boolean isRunning() {
		return started.get();
	}
}
